package com.zf.canal.demo;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * canal同步的简单样例，定时按照固定频率执行，没有使用spring
 */
public class SimpleCanalSyncClientDemo {

    private final static int BATCH_SIZE = 1000;

    public static void main(String[] args) {
        // 打开连接
        CanalConnector connector = openConnector();
        //创建一个定时执行器
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(4);
        executor.setMaximumPoolSize(8);
        // 定时同步数据， 在 上一次任务执行结束 到 下一次任务开始 之间 间隔 2s，
        // 注意是在上一个任务执行结束后才会开始等待执行下一次的任务执行，如果上一次任务没有结束，则不会开始下一次的等待
        executor.scheduleWithFixedDelay(() -> {
            try {
                // Message可以包含多个sql的执行结果, 每个sql的执行结果被封装成Entry，每个sql可能对多行数据造成影响，
                // 影响到行的集合被封装称RowChange，每行被封装成RowData, RowData包含变更前的数据和变更后的数据
                // 所以: Message 一个 -> 多个 Entry 一个 -> 一个 RowChange 一个 -> 多个 RowData 一个 -> 一个变更前和一个变更后数据
                // 获取指定条数的日志数据，具体数量以实际为准，小于等于 BATCH_SIZE，不阻塞，立即返回
                Message message = connector.getWithoutAck(BATCH_SIZE);
                //获取批次的ID
                long batchId = message.getId();
                //获取这个批次取到了多少数据
                int size = message.getEntries().size();
                try {
                    if (batchId != -1 && size > 0) {
                        //如果有数据, 就处理数据
                        for (Entry entry : message.getEntries()) {
                            // 只处理 ROWDATA 类型的 binlog
                            if (entry.getEntryType() == EntryType.ROWDATA) {
                                printEntry(entry);
                            }
                        }
                    }
                    //处理完成后，进行 batchId 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                    connector.ack(batchId);
                } catch (Exception e) {
                    System.out.printf("出现异常，Message：%s； 错误：%s", JSONObject.toJSONString(message), e.getMessage());
                    // 如果rollback，那么下次再次获取这些数据，还是有可能出错，造成死循环
                    // connector.rollback(batchId);
                    // 如果 ack, 那么这些操作产生的数据就会丢失，只能通知到人，手动处理
                    connector.ack(batchId);
                }
            } catch (Exception e) {
                System.out.println("canal定时同步任务异常");
                e.printStackTrace();
                connector.disconnect();
                executor.shutdown();
            }
        }, 0, 2, TimeUnit.SECONDS);
    }

    /**
     * 创建一个访问canal的连接
     */
    public static CanalConnector openConnector() {
        // destination 表示连接到canal的哪个实例上，默认启动的实例名为 example
        // 创建一个连接单个canal服务的连接, username和password 是连接canal的用户名密码，默认没有用户名和密码，可随便填写，
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
        //打开连接
        connector.connect();
        //订阅哪些数据库的哪些表的变化，正则表达式
        //如果本次订阅中filter信息为空，则直接使用canal server服务端配置的canal.instance.filter.regex 信息
        //如果本次订阅中filter信息不为空，目前会直接替换canal server服务端配置的canal.instance.filter.regex 信息，以本次提交的为准
        connector.subscribe("canal\\..*");
        //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
        connector.rollback();
        return connector;
    }

    /**
     * 打印canal server解析binlog获得的实体类信息
     */
    private static void printEntry(Entry entry) {
        //RowChange对象，包含了sql影响到的行数据，以及sql类型等相关信息
        RowChange rowChange;
        try {
            rowChange = RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            throw new RuntimeException("Entry 转换为 RowChange失败: " + entry.toString(), e);
        }
        //打印Header信息
        System.out.printf("binlog: %s, binlogOffset: %s, dataBaseName: %s, tableName: %s\n",
                entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
        //判断是否是DDL语句
        if (rowChange.getIsDdl()) {
            System.out.println("isDdl: true, sql: " + rowChange.getSql());
        }
        //获取sql操作类型：insert/update/delete/create等类型
        EventType eventType = rowChange.getEventType();
        System.out.println("操作类型：" + eventType);
        //获取RowChange对象里的每一行数据，只有增删改操作才会有 RowData 行数据，其他操作的getRowDatasList是空list
        for (RowData rowData : rowChange.getRowDatasList()) {
            // 只要有 RowData 行数据，那么 getBeforeColumnsList 和 getAfterColumnsList 一定是 >= 0 list，不会为null
            switch (eventType) {
                case DELETE:
                    //如果是删除语句
                    printColumn(rowData.getBeforeColumnsList());
                    break;
                case INSERT:
                    //如果是新增语句
                    printColumn(rowData.getAfterColumnsList());
                    break;
                case UPDATE:
                    //如果是更新的语句
                    //变更前的数据
                    System.out.println("更新前的数据：");
                    printColumn(rowData.getBeforeColumnsList());
                    //变更后的数据
                    System.out.println("更新后的数据：");
                    printColumn(rowData.getAfterColumnsList());
                    break;
            }
        }
    }

    /**
     * 打印一行中所有列的值
     */
    private static void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}
